package top.lunarye.api.action;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.google.common.base.Throwables;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import top.lunarye.api.domain.SendTaskModel;
import top.lunarye.common.domaim.TaskInfo;
import top.lunarye.common.enums.DelayType;
import top.lunarye.common.enums.RespStatusEnum;
import top.lunarye.common.vo.BasicResultVO;
import top.lunarye.support.delayMessage.DelayMessageProducer;
import top.lunarye.support.pipeline.BusinessProcess;
import top.lunarye.support.pipeline.ProcessContext;
import top.lunarye.support.utils.RocketMQUtils;

import javax.annotation.Resource;
import java.util.Iterator;

/**
 * 将消息发送到MQ
 * @author lunar
 * @since 2022-07-12
 */
@Service
@Slf4j
public class SendMQAction implements BusinessProcess<SendTaskModel> {

    @Resource
    private DelayMessageProducer delayMessageProducer;

    @Value("${message.business.topic.name}")
    private String topicName;

    @Override
    public void process(ProcessContext<SendTaskModel> context) {
        SendTaskModel sendTaskModel = context.getProcessModel();
        byte[] messageBytes = JSON.toJSONBytes(sendTaskModel.getTaskInfo(), JSONWriter.Feature.WriteClassName);

        TaskInfo taskInfo = CollUtil.getFirst(sendTaskModel.getTaskInfo().listIterator());
        Message message = new Message(topicName, messageBytes);
        try {
            if (taskInfo.getShieldType().equals(DelayType.DELAY_SEND.getCode())) {
                delayMessageProducer.sendDelay(message, taskInfo.getDelayTime());
            } else {
                delayMessageProducer.send(message);
            }
        } catch (Exception e) {
            context.setNeedBreak(true).setResponse(BasicResultVO.fail(RespStatusEnum.SERVICE_ERROR));
            log.error("send rocketMQ fail! e:{},params:{}", Throwables.getStackTraceAsString(e)
                    , JSON.toJSONString(CollUtil.getFirst(sendTaskModel.getTaskInfo().listIterator())));
        }
    }
}
